#' Make batched query to Tushare
#'
#' @param tbl tables to query
#' @param ... args passed to query
#' @param ts a tsapi object
#' @param timeout timeout
#' @param show_progress whether to show progress bar
#'
#' @return a list of query results and errors
#' @export
#'
ts_batch_query <- function(tbl, ..., ts = tsapi(), timeout = 10, show_progress = FALSE)
{
  # collect query parameters
  query <- data.table::CJ(..., unique = TRUE)
  query[, timeout := timeout]
  n_query <- nrow(query)

  # setup progress bar
  pb <- NULL
  if (show_progress)
  {
    pb <- progress::progress_bar$new(
      format = "Query [:bar] :percent eta: :eta",
      total = n_query * length(tbl)
    )
  }

  dt_list <- list()
  err_list <- list()
  for (t in tbl)
  {
    # generate query function
    f <- `$.tsapi`(ts, t)
    tmp_dt <- list()
    tmp_err_qidx <- c()
    for (i in seq_len(n_query))
    {
      # perform query
      tryCatch({
        dt <- do.call(f, query[i])
        if (nrow(dt))
        {
          tmp_dt[[i]] <- dt
        }
      },
      # collect error query
      error = function(e)
      {
        warning("Error making query: tbl ", t, ": ", e, call. = FALSE)
        tmp_err_qidx <<- c(tmp_err_qidx, i)
      })
      # update progress bar
      if (show_progress)
      {
        pb$tick()
      }
    }
    if (length(tmp_err_qidx))
    {
      err_list[[t]] <- query[tmp_err_qidx]
    }
    dt_list[[t]] <- data.table::rbindlist(tmp_dt)
  }

  list(
    result = dt_list,
    error = err_list
  )
}

#' Make batch query to Tushare and persist to database
#'
#' @param con a DBIConnection object
#' @param tbl tables to query, also written to con
#' @param ... args passed to query
#' @param ts a tsapi object
#' @param timeout timeout
#' @param show_progress whether to show progress bar
#' @param conflict passed to db_insert_to()
#' @param constraint passed to db_insert_to()
#'
#' @return a list of successful and failed queries
#' @export
#'
ts_batch_persist <- function(con, tbl, ..., ts = tsapi(), timeout = 10, show_progress = FALSE, conflict = "default", constraint = NULL)
{
  # collect query parameters
  query <- data.table::CJ(..., unique = TRUE)
  query[, timeout := timeout]
  n_query <- nrow(query)

  pb <- NULL
  # setup progress bar
  if (show_progress)
  {
    pb <- progress::progress_bar$new(
      format = "Query [:bar] :percent eta: :eta",
      total = n_query * length(tbl)
    )
  }

  query_list <- c()
  insert_list <- c()
  err_list <- list()
  for (t in tbl)
  {
    # generate query function
    f <- `$.tsapi`(ts, t)
    query_list[t] <- 0
    insert_list[t] <- 0
    tmp_err_qidx <- c()
    for (i in seq_len(n_query))
    {
      # perform query and write to con
      tryCatch({
        dt <- do.call(f, query[i])
        query_list[t] <- query_list[t] + nrow(dt)
        if (nrow(dt))
        {
          insert_list[t] <-  insert_list[t] + db_insert_to(con = con, tbl = t, dt = dt, conflict = conflict, constraint = constraint)
        }
      },
      # collect error query
      error = function(e)
      {
        warning("Error making query: tbl ", t, ": ", e, call. = FALSE)
        tmp_err_qidx <<- c(tmp_err_qidx, i)
      })
      # update progress bar
      if (show_progress)
      {
        pb$tick()
      }
    }
    if (length(tmp_err_qidx))
    {
      err_list[[t]] <- query[tmp_err_qidx]
    }
  }

  list(
    query = data.table::data.table(tbl = names(query_list), query = query_list, insert = insert_list),
    error = err_list
  )
}

#' Convert Tushare data date columns to IDate
#'
#' @param dt a data.table
#'
#' @return dt, setDT() and changed by ref
#' @export
#'
ts_conv_idate <- function(dt)
{
  data.table::setDT(dt)

  cols <- colnames(dt)
  col_date <- which(stringr::str_detect(cols, "date$|^period$"))
  if (length(col_date))
  {
    dt[, (col_date) := lapply(.SD, data.table::as.IDate), .SDcol = col_date]
  }

  dt
}

#' Convert Tushare data time columns to ITime
#'
#' @param dt a data.table
#' @param tz converted timezone, defaults to Asia/Shanghai
#'
#' @return dt, setDT() and changed by ref
#' @export
#'
ts_conv_itime <- function(dt, tz = "Asia/Shanghai")
{
  data.table::setDT(dt)

  conv_f <- function(x)
  {
    data.table::as.ITime(
      lubridate::as_datetime(x, tz = tz)
    )
  }

  cols <- colnames(dt)
  col_time <- which(stringr::str_detect(cols, "time$"))
  if (length(col_time))
  {
    dt[, (col_time) := lapply(.SD, conv_f), .SDcol = col_time]
  }

  dt
}

#' Convert Tushare data logical columns to logical values
#'
#' @param dt a data.table
#'
#' @return dt, setDT() and changed by ref
#' @export
#'
ts_conv_logical <- function(dt)
{
  data.table::setDT(dt)

  cols <- colnames(dt)
  col_logi <- which(stringr::str_detect(cols, "^is_|_flag$"))
  if (length(col_logi)) {
    dt[, (col_logi) := lapply(.SD, as.logical), .SDcol = col_logi]
  }
}
